package com.hadwinling.alogriithm.pagerank

import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object PageRank {
  def main(args: Array[String]): Unit = {
    val partitionsNum = 4 //设置电脑cpu的核心数量
    val conf = new SparkConf().setAppName("SparkPSO").setMaster("local")

    val sc = new SparkContext(conf)

    val links = sc.parallelize(List(("A",List("B","C")),("B",List("A","C")),("C",List("A","B","D")),("D",List("C"))),partitionsNum)
      .partitionBy(new HashPartitioner(100)).persist()

    var ranks=links.mapValues(v=>1.0)

    for (i <- 0 until 10) {
      val contributions=links.join(ranks).flatMap {
        case (pageId,(links,rank)) => links.map(dest=>(dest,rank/links.size))
      }
      ranks=contributions.reduceByKey((x,y)=>x+y).mapValues(v=>0.15+0.85*v)
    }

    ranks.sortByKey().collect().foreach(println)
  }
}
